use std::{
    alloc::System,
    io,
    num::NonZeroUsize,
    rc::Rc,
    sync::{
        atomic::{AtomicUsize, Ordering},
        Arc,
    },
    time::Duration,
};

use tokio::sync::{
    mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender},
    oneshot,
};

use crate::{
    service::InternalServiceFactory,
    socket::MioStream,
    waker_queue::{self, WakerQueue},
    Server,
};

pub(crate) struct Stop {
    graceful: bool,
    tx: oneshot::Sender<bool>,
}

pub(crate) struct Conn {
    pub io: MioStream,
    pub token: usize,
}
#[derive(Clone)]
pub(crate) struct Counter {
    counter: Arc<AtomicUsize>,
    limit: usize,
}

impl Counter {
    pub(crate) fn new(limit: usize) -> Self {
        Self {
            counter: Arc::new(AtomicUsize::new(1)),
            limit,
        }
    }
    // 累加
    pub(crate) fn inc(&self) -> bool {
        self.counter.fetch_add(1, Ordering::Relaxed) != self.limit
    }

    // 累减，当到达 limit 时为 true
    pub(crate) fn dec(&self) -> bool {
        self.counter.fetch_sub(1, Ordering::Relaxed) == self.limit
    }

    // 考虑为什么要使用 Ordering::SeqCst ，这是最严格的
    pub(crate) fn total(&self) -> usize {
        self.counter.load(Ordering::SeqCst) - 1
    }
}

pub(crate) struct WorkerCounter {
    idx: usize,
    inner: Rc<(WakerQueue, Counter)>,
}

impl Clone for WorkerCounter {
    fn clone(&self) -> Self {
        Self {
            idx: self.idx,
            inner: self.inner.clone(),
        }
    }
}

impl WorkerCounter {
    pub(crate) fn new(idx: usize, waker_queue: WakerQueue, counter: Counter) -> Self {
        Self {
            idx,
            inner: Rc::new((waker_queue, counter)),
        }
    }

    pub(crate) fn guard(&self) -> WorkerCounterGuard {
        WorkerCounterGuard(self.clone())
    }

    fn total(&self) -> usize {
        self.inner.1.total()
    }
}

pub(crate) struct WorkerCounterGuard(WorkerCounter);

pub(crate) struct WorkerHandleAccept {
    idx: usize,
    conn_tx: UnboundedSender<Conn>,
    counter: Counter,
}

impl WorkerHandleAccept {
    pub(crate) fn idx(&self) -> usize {
        self.idx
    }

    pub(crate) fn send(&self, conn: Conn) -> Result<(), Conn> {
        // 成功不处理，错误返回Conn
        self.conn_tx.send(conn).map_err(|msg| msg.0)
    }

    pub(crate) fn inc_counter(&self) -> bool {
        self.counter.inc()
    }
}

pub(crate) struct WorkerHandleServer {
    pub(crate) idx: usize,
    stop_tx: UnboundedSender<Stop>,
}

impl WorkerHandleServer {
    pub(crate) fn stop(&self, graceful: bool) -> oneshot::Receiver<bool> {
        let (tx, rx) = oneshot::channel();
        let _ = self.stop_tx.send(Stop { graceful, tx });
        rx
    }
}
#[derive(Debug, Clone, Copy)]
pub(crate) struct ServerWorkerConfig {
    shutdown_timeout: Duration,
    max_blocking_threads: usize,
    max_concurrent_connections: usize,
}

impl Default for ServerWorkerConfig {
    fn default() -> Self {
        let parallelism = std::thread::available_parallelism().map_or(2, NonZeroUsize::get);
        let max_blocking_threads = std::cmp::max(512 / parallelism, 1);
        Self {
            shutdown_timeout: Duration::from_secs(30),
            max_blocking_threads,
            max_concurrent_connections: 25600,
        }
    }
}

enum WorkerServiceStatus {
    Available,
    Unavailable,
    Failed,
    Restarting,
    Stopping,
    Stopped,
}

struct WorkerService {
    factory_idx: usize,
    status: WorkerServiceStatus,
}
/// worker accepts Socket objects via unbounded channel and starts stream processing.
pub(crate) struct ServerWorker {
    // UnboundedReceiver<Conn> should always be the first field.
    // It must be dropped as soon as ServerWorker dropping.
    // 所以drop 是按字段顺序进行drop 么？
    conn_rx: UnboundedReceiver<Conn>,
    stop_rx: UnboundedReceiver<Stop>,
    counter: WorkerCounter,
    factories: Box<[Box<dyn InternalServiceFactory>]>,
}

impl ServerWorker {
    pub(crate) fn start(
        idx: usize,
        factories: Vec<Box<dyn InternalServiceFactory>>,
        waker_queue: WakerQueue,
        config: ServerWorkerConfig,
    ) -> io::Result<(WorkerHandleAccept, WorkerHandleServer)> {
        let (tx1, conn_rx) = unbounded_channel();
        let (tx2, stop_rx) = unbounded_channel();

        let counter = Counter::new(config.max_concurrent_connections);
        let pair = handle_pair(idx, tx1, tx2, counter.clone());

        // TODO: 待实现
        Ok(pair)
    }
}

fn handle_pair(
    idx: usize,
    conn_tx: UnboundedSender<Conn>,
    stop_tx: UnboundedSender<Stop>,
    counter: Counter,
) -> (WorkerHandleAccept, WorkerHandleServer) {
    let accept = WorkerHandleAccept {
        idx,
        conn_tx,
        counter,
    };

    let server = WorkerHandleServer { idx, stop_tx };
    (accept, server)
}
